package com.qf.consumer.listeners;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.qf.consumer.pojo.User;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;

@Component
public class RabbitMQListener {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public static final String RETRIES_COUNT_PREFIX = "consumer:";

//    @RabbitListener(queues = "boot.que1")
    public void consumer1(Message msg){
        byte[] body = msg.getBody();
        System.out.println("consumer1--msg:" + new String(body));
        System.out.println("---------------------");
    }

//    @RabbitListener(queues = "boot.que1")
    public void consumer2(Message msg, Channel channel) throws IOException {
        try {
            byte[] body = msg.getBody();
            String json = new String(body);
            System.out.println("consumer2--msg:" + new ObjectMapper().readValue(json, User.class));

            //模拟异常
//            int i = 1/0;

            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),true);
            System.out.println("---------------------");
        }catch (Exception e){
            e.printStackTrace();

            String key = RETRIES_COUNT_PREFIX + msg.getMessageProperties().getMessageId();
            Long retries = redisTemplate.opsForValue().increment(key);

            if (retries <= 2) {
                //重新入队，重试消费
                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), true, true);
            }else{
                //不如对，丢弃(如果有死信队列，进入死信中)
                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), true, false);
            }
        }
    }

//    @RabbitListener(queues = "boot.que1")
//    public void consumer3(User user){
//        System.out.println("consumer2--msg:" + user);
//        System.out.println("---------------------");
//    }

//    @RabbitListener(queues = "boot.normalQue")
    public void consumer3(Message msg, Channel channel) throws IOException {
        //不确认并且不入队，消息转为死信
        channel.basicNack(msg.getMessageProperties().getDeliveryTag(),true,false);
    }

//    @RabbitListener(queues = "boot.deadQue",ackMode = "AUTO")
//    public void consumer4(Message msg) throws IOException {
//        //不确认并且不入队，消息转为死信
//        byte[] body = msg.getBody();
//        System.out.println("msg:" + new String(body) + "-- time:" + LocalDateTime.now());
//    }

    /*
        演示幂等性问题
        增：非幂等
            业务流程：① 查询这个数据是否存在 ②不存在，数据做插入动作
            解决方案：
                ① 利用分布式锁保障上述业务操作的原子性 （redis： setnx/redisson）
                ② 利用数据库的悲观锁实现 （不推荐使用）
                    select ... from 表名 where 条件 for update  --  for update 给查询语句上写锁
                    insert into ...    -- 自动申请写锁
                    *** 注意：上述两个动作必须在同一个事务中
                ③ 利用数据库的唯一约束
                    业务： 数据做插入动作，数据中的有一个唯一约束的数据--用消息id
                     *** 注意：对唯一约束异常进行捕获处理
                ④ 去重表
                    将唯一约束的数据额外建一个去重表，将业务数据和去重表中的唯一约束字段分离
                    当我们向业务表中添加数据时，需要同步向去重表添加数据，并且这两个添加动作必须在一个事物中

        删：幂等

        改：
            设置型：幂等
                update xx set 列=值,列2=值2,... where ...
            计算型：非幂等
                update xx set 列=列+值,列2=列2+值2,... where ...

                解决方案：
                    ⑤ 状态机，利用额外的一个状态字段标识这条记录是否被修改过
                    update xx set 列=列+值,列2=列2+值2,status='已修改'... where ... and status='未修改'

        查：幂等

     */

    public static final String MD_PREFIX = "MD:";

    @RabbitListener(queues = "boot.normalQue",ackMode = "AUTO")
    public void consumer5(Message msg) throws IOException {

        String key = MD_PREFIX + msg.getMessageProperties().getMessageId();
        Boolean b = redisTemplate.opsForValue().setIfAbsent(key, "");

        if (b) {
            try {
                //不确认并且不入队，消息转为死信
                byte[] body = msg.getBody();
                System.out.println("msg:" + new String(body) + "--msgId:" + msg.getMessageProperties().getMessageId());
            }catch (Exception e){
                //业务执行失败，需要进行重试，必须将redis中该key删除
                redisTemplate.delete(key);
            }
        }
    }
}
